package com.you.raft.core.task;

import com.you.raft.common.entity.*;
import com.you.raft.core.LogModule;
import com.you.raft.core.StateMachine;
import com.you.raft.core.impl.DefaultRaftNode;
import com.you.raft.rpc.RpcRaftClient;
import com.you.raft.common.entity.Request;
import com.you.raft.thread.RaftThreadPool;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 1. 在转变成CANDIDATE后就立即开始选举过程
 * 自增当前的任期号（currentTerm）
 * 给自己投票
 * 重置选举超时计时器
 * 发送请求投票的 RPC 给其他所有服务器
 * 2. 如果接收到大多数服务器的选票，那么就变成领导人
 * 3. 如果接收到来自新的领导人的附加日志 RPC，转变成跟随者
 * 4. 如果选举过程超时，再次发起一轮选举
 */
@Slf4j
public class ElectionTask implements Runnable {

    private DefaultRaftNode node;
    private LogModule logModule;
    private StateMachine stateMachine;
    private Cluster cluster;
    private RpcRaftClient rpcClient;
    /**
     * 选举时间间隔基数
     */
    private static final long electionTime = 10 * 1000;

    public ElectionTask(DefaultRaftNode node) {
        this.node = node;
        this.logModule = node.getLogModule();
        this.stateMachine = node.getStateMachine();
        this.cluster = node.getCluster();
        this.rpcClient = node.getRpcClient();
    }

    @Override
    public void run() {
        if (node.getStatus() == NodeStatus.LEADER) {
            return;
        }
        //判断当前时间与上一次选举的时间间隔是否合理的时间范围内
        if (System.currentTimeMillis() - node.getPreElectionTime() < electionTime + ThreadLocalRandom.current().nextInt(150) + 150) {
            return;
        }
        node.setPreElectionTime(System.currentTimeMillis());
        node.setStatus(NodeStatus.CANDIDATE);
        log.info("当前节点{}状态变为【CANDIDATE】,开始一轮选举,当前任期:[{}],最后的日志项:[{}]",
                cluster.getSelf(), node.getCurrentTerm(), logModule.getLast());
        //增加任期号
        node.setCurrentTerm(node.getCurrentTerm() + 1);
        //投票给自己
        node.setVotedFor(cluster.getSelf().getAddress());
        //获取其他节点的投票
        List<Node> clusterOtherNodes = cluster.getOtherNodes();
        List<Future<VoteResponse>> voteRequestTask = new ArrayList<>();
        for (Node cnode : clusterOtherNodes) {
            voteRequestTask.add(RaftThreadPool.submit(() -> {
                long lastTerm = 0L;
                LogEntry last = logModule.getLast();
                if (last != null) {
                    lastTerm = last.getTerm();
                }
                VoteRequest param = VoteRequest.builder().
                        term(node.getCurrentTerm()).
                        candidateId(cluster.getSelf().getAddress()).
                        lastLogIndex(node.getLogModule().getLastIndex()).
                        lastLogTerm(lastTerm).
                        build();

                Request request = Request.builder()
                        .cmd(Request.VOTE)
                        .obj(param)
                        .addr(cnode.getAddress())
                        .build();
                return rpcClient.send(request);
            }));
        }
        AtomicInteger ticket = new AtomicInteger(0);
        CountDownLatch latch = new CountDownLatch(voteRequestTask.size());
        // 获取投票结果
        for (Future<VoteResponse> future : voteRequestTask) {
            RaftThreadPool.submit(() -> {
                try {
                    VoteResponse result = future.get(3000, TimeUnit.MILLISECONDS);
//                    log.info("处理响应:{}", result);
                    if (result == null) {
                        return -1;
                    }
                    if (result.isVoteGranted()) {
                        ticket.incrementAndGet();
                    } else {
                        // 更新自己的任期.
                        long resTerm = result.getTerm();
                        if (resTerm >= node.getCurrentTerm()) {
                            node.setCurrentTerm(resTerm);
                        }
                    }
                    return 0;
                } catch (Exception e) {
                    return -1;
                } finally {
                    latch.countDown();
                }
            });
        }
        try {
            latch.await(4000, TimeUnit.MICROSECONDS);
        } catch (InterruptedException e) {
        }
        int success = ticket.get();
        log.info("节点 {} 得到票数: {} / {} , 状态: {}", cluster.getSelf(), success, voteRequestTask.size(),
                NodeStatus.Enum.value(node.getStatus()));
        // 如果投票期间,有其他服务器发送 AppendEntry , 就可能变成 follower ,这时,应该停止.
        if (node.getStatus() == NodeStatus.FOLLOWER) {
            log.info("节点 {} 于选举过程中状态变为了【FOLLOWER】", cluster.getSelf());
            node.setVotedFor("");
            return;
        }
        if (success >= (clusterOtherNodes.size() + 1) / 2) {
            log.warn("节点 {} 成为 【LEADER】 ", cluster.getSelf());
            node.setStatus(NodeStatus.LEADER);
            cluster.setLeader(cluster.getSelf());
            //成为Leader
            becomeLeaderToDoThing();
        }
        // 再次更新选举时间
        node.setVotedFor("");
        node.setPreElectionTime(System.currentTimeMillis());
    }

    /**
     * 初始化所有的 nextIndex 值为自己的最后一条日志的 index + 1.
     * 如果下次 RPC 时, 跟随者和leader 不一致,就会失败.
     * 那么 leader 尝试递减 nextIndex 并进行重试.最终将达成一致.
     */
    private void becomeLeaderToDoThing() {
        //重新初始化nextIndexs与matchIndex数组
        Map<Node, Long> nextIndexs = new ConcurrentHashMap<>();
        Map<Node, Long> matchIndexs = new ConcurrentHashMap<>();
        for (Node cnode : cluster.getOtherNodes()) {
            //下一个发送的日志索引:在replication会根据节点的nextIndex
            nextIndexs.put(cnode, logModule.getLastIndex() + 1);
            //已经复制完成的日志索引
            matchIndexs.put(cnode, 0L);
        }
        node.setNextIndexs(nextIndexs);
        node.setMatchIndexs(matchIndexs);
        // 创建[空日志]并提交-用于处理当前节点之前未提交的日志
        LogEntry logEntry = LogEntry.builder()
                .command(null)
                .term(node.getCurrentTerm())
                .index(-1).build();
        // 提交到本地日志-日志项logEntry的索引会在这个方法中更新
        logModule.write(logEntry);
        log.info("日志: {}, 日志索引 : {}", logEntry, logEntry.getIndex());
        // 将空日志复制到其他机器
        List<Future<Boolean>> taskList = new ArrayList<>();
        for (Node cnode : cluster.getOtherNodes()) {
            taskList.add(node.replication(cnode, logEntry));
        }
        int taskCount = taskList.size();
        final AtomicInteger sucCount = new AtomicInteger(0);
        CountDownLatch latch = new CountDownLatch(taskCount);
        for (Future<Boolean> future : taskList) {
            RaftThreadPool.execute(() -> {
                try {
                    if (future.get(3000, TimeUnit.MILLISECONDS)) {
                        sucCount.incrementAndGet();
                    }
                } catch (Exception e) {
                    log.error("错误:", e);
                } finally {
                    latch.countDown();
                }
            });
        }
        try {
            latch.await(4000, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            log.error("错误:", e);
        }
        // 如果存在一个满足N > commitIndex的 N，并且大多数的matchIndex[i] ≥ N成立，
        // 并且log[N].term == currentTerm成立，那么令 commitIndex 等于这个 N
        List<Long> matchIndexList = new ArrayList<>(matchIndexs.values());
        int median = 0;
        if (matchIndexList.size() >= 2) {
            Collections.sort(matchIndexList);
            median = matchIndexList.size() / 2;
        }
        long N = matchIndexList.get(median);
        if (N > node.getCommitIndex()) {
            LogEntry entry = logModule.get(N);
            if (entry != null && entry.getTerm() == node.getCurrentTerm()) {
                node.setCommitIndex(N);
            }
        }
        //  响应客户端成功一半，应用日志到状态机
        if (sucCount.get() >= (taskCount + 1) / 2) {
            node.setCommitIndex(logEntry.getIndex());
            stateMachine.apply(logEntry);
            node.setLastApplied(node.getCommitIndex());
            log.info("日志成功应用到状态机 : {}", logEntry);
        } else {
            // 回滚已经提交的日志
            logModule.removeStartWith(logEntry.getIndex());
            log.warn("日志提交失败: {}", logEntry);
            // 无法提交空日志，让出领导者位置
            log.warn("节点 {} 领导者下线,变为【FOLLOWER】", cluster.getSelf());
            node.setStatus(NodeStatus.FOLLOWER);
            cluster.setLeader(null);
        }
    }
}
